package com.shujia.opt

import org.apache.spark.{Partitioner, SparkContext}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object Code02SparkPartitionerData {
  def main(args: Array[String]): Unit = {
    /**
     * Spark数据倾斜：
     * 表现：在Job任务执行时，部分的Task一直处于运行状态，而其他的Task已经处理完成
     *
     * 方式2：自定义分区器，将产生数据倾斜的Key 单独分别放入不同的分区中进行处理 其他剩余数据，直接分配给不同的分区中
     *
     */


    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("spark")
      .getOrCreate()

    val sc: SparkContext = spark.sparkContext
    val data: RDD[String] = sc.textFile("spark_code/data/word/*")

    data
      .flatMap(_.split(","))
      .map((_, 1))
      .groupBy(_._1, new Partitioner {
        override def numPartitions: Int = 3

        override def getPartition(key: Any): Int = {
          val keyStr: String = key.toString
          keyStr match {
            case "NULL" => 0
            case key: String => {
              (key.hashCode % 2) + 1
            }
          }
        }
      })
      .mapValues(_.size)
      .foreach(println)


    // (NULL,23680)
    while (true) {}
  }
}
